package avroserialization;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import org.apache.avro.Schema;
import org.apache.avro.file.DataFileStream;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.specific.SpecificRecord;

/**
 *
 * @author mwh
 *
 * Mulige optimeringer 1. Ikke enum, men objevt der muliggør genbrug af encoders
 * og decoders 2. Genbrug af SpecificRecord ved at give som input til metode
 *
 * Andet: 1. Lav Generic så der ikke skal castes
 */
public enum AvroSerialization {

    INSTANCE;

    public void serializeWithSchema(OutputStream os, SpecificRecord record) throws IOException {
        DatumWriter<SpecificRecord> datumWriter = new SpecificDatumWriter<SpecificRecord>(record.getSchema());
        DataFileWriter<SpecificRecord> fileWriter = new DataFileWriter<SpecificRecord>(datumWriter);
        fileWriter.create(record.getSchema(), os);
        fileWriter.append(record);
        fileWriter.close();
    }

    public void serializeWithoutSchema(OutputStream os, SpecificRecord record) throws IOException {
        DatumWriter<SpecificRecord> datumWriter = new SpecificDatumWriter<SpecificRecord>(record.getSchema());
        BinaryEncoder becoder = EncoderFactory.get().directBinaryEncoder(os, null);
        datumWriter.setSchema(record.getSchema());
        datumWriter.write(record, becoder);
        becoder.flush();
    }

    public SpecificRecord desirializeWithSchema(InputStream is, Schema schema) throws IOException {
        DatumReader<SpecificRecord> datumReader = new SpecificDatumReader<SpecificRecord>(schema);
        DataFileStream<SpecificRecord> dataReader = new DataFileStream<SpecificRecord>(is, datumReader);
        return dataReader.next();
    }

    public SpecificRecord desirializeWithoutSchema(InputStream is, Schema schema) throws IOException {
        DatumReader<SpecificRecord> datumReader = new SpecificDatumReader<SpecificRecord>(schema);
        BinaryDecoder bdecoder = DecoderFactory.get().directBinaryDecoder(is, null);
        datumReader.setSchema(schema);
        return datumReader.read(null, bdecoder);
    }
}
